package gspushmq.v1.produce.job.sys;

import java.io.IOException;
import java.nio.charset.Charset;

import org.apache.commons.lang3.StringUtils;

import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.MessageProperties;

import ggframework.util.GGConfigurer;
import gspushmq.model.GGQueueConstant;
import gspushmq.util.GGLogger;
import gspushmq.v1.produce.job.core.GGRabbitMQ;
import gspushmq.v1.produce.job.hq.HQService;

/**
 * @Description 实时获取mq中的消息入库
 * @Author chengsj
 * @Date 2018/11/29 11:09
 * @Version 1.0
 * @company 上海朝阳永续信息技术有限公司
 * @copyright (c) 2015 SunTime Co'Ltd Inc. All rights reserved.
 **/
public class GGMQRealTimeJob {


    private static final ThreadLocal<Channel> THREAD_LOCAL = new ThreadLocal<Channel>();


    /**
     * 投递消息
     *
     * @param queueName 队列名
     * @param message   消息
     *                  Created by chengsj on 2018/11/29
     */
    public static void sendMQ(String queueName, String message) {
        if (StringUtils.isBlank(queueName) || StringUtils.isBlank(message)) {
            throw new NullPointerException("队列名、数据流不能为空");
        }
        Channel channel = null;
        try {
            channel = GGRabbitMQ.GG_RABBIT_MQ.getConnection().createChannel();
            String exchangeName = queueName;
            String exchangeType = "direct";
            channel.exchangeDeclare(exchangeName, exchangeType, true);
            channel.basicPublish(exchangeName, queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(Charset.forName("UTF-8")));
            GGLogger.info(GGMQRealTimeJob.class, "---> 消息投递" + queueName + "成功!rootkey=" + queueName + " ; msg=" + message+"| channel="+channel.getConnection());
        } catch (Exception e) {
            GGLogger.error(GGMQRealTimeJob.class, e, "消息队列异常，队列名：%s，数据流：%s", queueName, message);
            throw new RuntimeException("消息发送异常");
        } finally {
            if (channel != null && channel.isOpen()) {
                try {
                    channel.close();
                } catch (IOException e) {
                    GGLogger.error(GGMQRealTimeJob.class, e, "消息队列通道关闭异常");
                }
            }
        }
    }

    /**
     * 获取mq通道,若没有获取到则创建一个
     * Created by chengsj on 2018/11/29
     */
    private static Channel getChannel() {
        Channel channel = THREAD_LOCAL.get();
        if (channel == null) {
            try {
                channel = GGRabbitMQ.GG_RABBIT_MQ.getConnection().createChannel();
                if (channel != null) {
                    THREAD_LOCAL.set(channel);
                }
            } catch (IOException e) {
                GGLogger.error(GGMQRealTimeJob.class, e, "");
            }
        }
        return channel;
    }


    /**
     * 消费消息
     * Created by chengsj on 2018/11/29
     */
    public static void consume(final String queueName) {
        GGLogger.info(GGMQRealTimeJob.class, "----> 队列" + queueName + "初始化开始-->");
        if (queueName == null) {
            throw new NullPointerException("队列名");
        }
        String queueKey = null;
        try {
            //消费时，一个线程保持一个
            final Channel channel = getChannel();
            //每次消费一个消息，处理完才消费第二条消息
            if (GGConfigurer.getBoolean("ggmes.basicQos", true)) {
                channel.basicQos(1);
            }
            //队列名称
            queueKey = channel.queueDeclare().getQueue();
            //交换机名称
            String exchangeName = queueName;
            //声明一个交换机
            channel.exchangeDeclare(exchangeName, "direct", true);
            //队列和交换机根据rootKey绑定
            channel.queueBind(queueKey, exchangeName, queueName);
            GGLogger.info(GGMQRealTimeJob.class,
                    new StringBuffer()
                            .append("---->链接：").append(channel.getConnection())
                            .append("\n---->队列").append(queueKey)
                            .append("\n---->交换机：").append(exchangeName)
                            .append("\n---->rootKey: ").append(queueName)
                            .append(" \n---->等待消息中....").toString());
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope,
                                           AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String message = new String(body, "UTF-8");
                    GGLogger.info(GGMQRealTimeJob.class, "---> 队列" + queueName + "消息接收成功！msg=" + message);
                    try {
                        if (StringUtils.isNotBlank(message)) {
                            JSONObject map_message = JSONObject.parseObject(message);
                            if (!map_message.isEmpty()) {
                                String mesType = map_message.getString(GGQueueConstant.MESSAGE_TYPE);
                                GGLogger.info(GGMQRealTimeJob.class, "消费者接收队列" + queueName + "的消息消息类型是：" + mesType + "：" + map_message.toJSONString());
                                //10001【股票预警消息】
                                if ("10001".equals(mesType)) {
                                    HQService.receiveHqMessage(map_message);
                                }
                            }
                        }
                    } catch (Exception e) {
                        GGLogger.error(GGMQRealTimeJob.class, "接收到消息格式转换异常" + e);
                    }
                }
            };
            //消费消息但立即删除
            channel.basicConsume(queueKey, true, consumer);
        } catch (Exception e) {
            GGLogger.error(GGMQRealTimeJob.class, e, "消息队列消费异常，队列名：%s", queueKey);
            throw new RuntimeException("消息消费异常");
        }
    }


}

